/**
 * Copyright (C) 2010, 2011 Neofonie GmbH
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

REGISTER $JAR

SET job.name 'Generic Twitter Counter from $STARTROW'

%default QUEUE_NAME dicode
SET mapreduce.job.queuename $QUEUE_NAME

%default DEFAULT_PARALLEL 20
SET default_parallel $DEFAULT_PARALLEL

%default MAX_FIELD_VALUE_LENGTH 100  -- maximum number of characters a value key may have
%default MIN_COUNT 3                 -- minimum count for a value to be included
%default TOP_FIELDS_NUMBER 100       -- number of top fields that is stored, rest is discarded


DEFINE splitIntoBag eu.dicodeproject.analysis.pig.SplitIntoBag('#');
DEFINE toJson eu.dicodeproject.analysis.pig.ResultToJson();

DEFINE storeCounts(dataset, field, OUTPUT_TABLE, MAX_FIELD_VALUE_LENGTH, MIN_COUNT, TOP_FIELDS_NUMBER) RETURNS void {
  filtered = FILTER $dataset BY
        SIZE($field) <= $MAX_FIELD_VALUE_LENGTH
    AND $field IS NOT NULL
    AND $field != '';

  grouped = GROUP filtered BY (day, $field);
  counted = FOREACH grouped GENERATE
    FLATTEN(group) AS (day:chararray, val:chararray),
    COUNT(filtered) AS cnt;

  thresholded = FILTER counted BY
    cnt >= $MIN_COUNT;
  withRowKey = FOREACH thresholded GENERATE
    LOWER(CONCAT('$field', CONCAT('_', day))) AS rowKey:chararray,
    val,
    cnt;

  groupedByRowKey = GROUP withRowKey BY rowKey;
  jsons = FOREACH groupedByRowKey {
    sorted = ORDER withRowKey BY cnt DESC;
    limited = LIMIT sorted $TOP_FIELDS_NUMBER;
    GENERATE
      group AS rowKey:chararray,
      toJson(limited) AS json:chararray;
  };

  STORE jsons INTO '$OUTPUT_TABLE'
    USING org.apache.pig.backend.hadoop.hbase.HBaseStorage('d:result', '-caster HBaseBinaryConverter');
};


tweets = LOAD '$INPUT_TABLE'
  USING org.apache.pig.backend.hadoop.hbase.HBaseStorage(
    'd:hashtags d:lang d:placeCountryCode d:placeFullName d:url d:from d:source d:inReplyToStatusId d:placeType d:to d:userIdsMentioned d:userNamesMentioned',
    '-loadKey true -gt $STARTROW')
  AS (id:chararray, hashtagsStr:chararray, lang:chararray, placeCountryCode:chararray, placeFullName:chararray, url:chararray, from:chararray, source:chararray, inReplyToStatusId:chararray, placeType:chararray, to:chararray, userIdsMentioned:chararray, userNamesMentioned:chararray);

tweetsWithDays = FOREACH tweets GENERATE
  id,
  SUBSTRING(id, 0, 8) AS day,
  hashtagsStr,
  lang,
  placeCountryCode,
  placeFullName,
  url;
  --placeType,
  --source,
  --inReplyToStatusId,
  --from,
  --to,
  --FLATTEN(splitIntoBag(userIdsMentioned)) AS userIdMentioned,
  --FLATTEN(splitIntoBag(userNamesMentioned)) AS userNameMentioned;

-- name of the field (e.g. hashtags, lang, placeFullName) corresponds (lower-cased) to rowKey prefix and is therefore important!

flatHashtags = FOREACH tweetsWithDays GENERATE
  day,
  FLATTEN(splitIntoBag(hashtagsStr)) AS hashtags:chararray;

storeCounts(flatHashtags, 'hashtags', '$OUTPUT_TABLE', '$MAX_FIELD_VALUE_LENGTH', '$MIN_COUNT', '$TOP_FIELDS_NUMBER');
storeCounts(tweetsWithDays, 'lang', '$OUTPUT_TABLE', '$MAX_FIELD_VALUE_LENGTH', '$MIN_COUNT', '$TOP_FIELDS_NUMBER');
storeCounts(tweetsWithDays, 'placeCountryCode', '$OUTPUT_TABLE', '$MAX_FIELD_VALUE_LENGTH', '$MIN_COUNT', '$TOP_FIELDS_NUMBER');
storeCounts(tweetsWithDays, 'placeFullName', '$OUTPUT_TABLE', '$MAX_FIELD_VALUE_LENGTH', '$MIN_COUNT', '$TOP_FIELDS_NUMBER');
storeCounts(tweetsWithDays, 'url', '$OUTPUT_TABLE', '$MAX_FIELD_VALUE_LENGTH', '$MIN_COUNT', '$TOP_FIELDS_NUMBER');
